package cn.smileyan.demos;

import cn.smileyan.demos.entity.Student;
import cn.smileyan.demos.io.CommonKafkaBuilder;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.utils.MultipleParameterTool;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * flink 任务入口
 * @author smileyan
 */
public class FlinkKafkaSourceDemo {
    /**
     * 参数解释：
     *  -bs broker 地址 localhost:9092
     *  -kcg kafka consumer group
     *  -it kafka 输入数据 topic test-input-topic
     *  -ot kafka 输出数据 topic test-output-topic
     *  -ct 可选，是否自动创建 topic，使用方法 添加  -ct 即可，无需指定其值
     *  -pt topic 可选，分区数 1
     *  -rf topic 可选，副本数 1
     *  -gs gapSeconds 可选，单个消息窗口间隔时间，默认 1000 ms
     *  example:
     *  -bs localhost:9092 -it test-input-topic -ot test-output-topic -pt 1 -rf 1 -ct
     */
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        final MultipleParameterTool parameterTool = MultipleParameterTool.fromArgs(args);
        final CommonKafkaBuilder kafkaBuilder = new CommonKafkaBuilder(parameterTool);
        final KafkaSource<Student> kafkaSource = kafkaBuilder.buildSource(Student.class);

        final DataStreamSource<Student> sourceStream = env.fromSource(kafkaSource,
                WatermarkStrategy.noWatermarks(), "Kafka Source");
        sourceStream.print();
        env.execute();
    }
}
